/******************************************************************************
 * Copyright (C) 2015  671643387@qq.com
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the
 * Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
 * more details.
 *
 * You should have received a copy of the GNU General Public License along
 * with this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 *
 *****************************************************************************/

#include "thread_pool.h"
#include "node/sys/lock_guard.h"
#include "node/sys/console.h"
#include <cstring>
using Console = ::node::sys::Console;

#define __NODE_LOCK(x)  ::node::sys::LockGuard<::node::sys::Mutex> guard(x)

namespace node {
    namespace sys {

        ThreadPool::ThreadPool(void)
            : init_thread_num_(0)
            , cur_thread_num_(0)
            , waiting_for_stop_(false)
            , stopping_(false)
            , task_queue_lock_()
            , task_queue_()
            , task_queue_event_()
        {
            memset(workers_, 0, 
                sizeof(::node::sys::ThreadPool::PoolThread *) * MAX_WORKERS);
        }

        ThreadPool::~ThreadPool(void)
        {
            Stop();
        }

        bool ThreadPool::Start(int thread_num)
        {
            init_thread_num_ = thread_num;
            do 
            {
                for (int x = 0; x < thread_num; ++x)
                {
                    ::node::sys::ThreadPool::PoolThread *worker 
                        = new ::node::sys::ThreadPool::PoolThread(*this);
                    if (worker && worker->Start()) {
                        workers_[x] = worker;
                        ++cur_thread_num_;
                        continue;
                    }
                    delete worker;
                }
                if (cur_thread_num_ == 0) {
                    return false;
                }
                if (cur_thread_num_ != init_thread_num_) {
                    Console::Warning("ThreadPool::Start: init_thread[%d], cur_thread[%d]",
                        init_thread_num_, cur_thread_num_);
                }
            } while (false);
            return true;
        }

        void ThreadPool::Stop(void)
        {
            if (stopping_) {
                return;
            }

            stopping_ = true;

            // clear all task
            {
                __NODE_LOCK(task_queue_lock_);
                while (!task_queue_.empty())
                {
                    ::node::sys::ThreadPool::Task* task = task_queue_.front();
                    task_queue_.pop_front();
                    task->complete_callback();
                    delete task;
                }
            }

            // wait all workers
            task_queue_event_.Set();
            for (int x = 0; x < cur_thread_num_; ++x) {
                if (workers_[x] != NULL) {
                    workers_[x]->Join();
                    delete workers_[x];
                    workers_[x] = NULL;
                }
            }
        }

        void ThreadPool::WaitAllTaskCompletedAndStop(void)
        {
            if (waiting_for_stop_) {
                return;
            }

            waiting_for_stop_ = true;

            // wait all task complete
            while (true)
            {
                __NODE_LOCK(task_queue_lock_);
                if (task_queue_.empty()) {
                    break;
                }
                ::node::sys::Thread::Sleep(300);
            }

            Stop();
        }

        ::node::sys::ThreadPool::Task* ThreadPool::NewTask(void)
        {
            if (stopping_ || waiting_for_stop_) {
                return NULL;
            }

            ::node::sys::ThreadPool::Task* task = new ::node::sys::ThreadPool::Task;
            if (task != NULL) {
                memset(task, 0, sizeof(::node::sys::ThreadPool::Task));
            }
            return task;
        }

        void ThreadPool::FreeTask(::node::sys::ThreadPool::Task * task)
        {
            assert(task != NULL);
            delete task;
        }

        bool ThreadPool::SubmitTask(::node::sys::ThreadPool::Task *task)
        {
            assert(task != NULL);

            if (stopping_ || waiting_for_stop_) {
                return false;
            }
            
            {
                __NODE_LOCK(task_queue_lock_);
                task_queue_.push_back(task);
            }
            task_queue_event_.Set();
            return true;
        }

        ::node::sys::ThreadPool::Task* ThreadPool::GetTask(void)
        {
            volatile ::node::sys::ThreadPool::Task* task = NULL;
            do 
            {
                volatile bool waiting = false;
                {
                    __NODE_LOCK(task_queue_lock_);
                    if (task_queue_.empty()) {
                        waiting = true;
                    }
                    else {
                        task = task_queue_.front();
                        task_queue_.pop_front();
                    }
                }
                if (waiting) {
                    task_queue_event_.Wait();
                }
            } while (task == NULL && !stopping_);
            return (::node::sys::ThreadPool::Task*)task;
        }

        ThreadPool::PoolThread::PoolThread(::node::sys::ThreadPool &owner)
            : owner_(owner)
        {

        }

        ThreadPool::PoolThread::~PoolThread(void)
        {

        }

        void ThreadPool::PoolThread::Stop(void)
        {

        }

        void ThreadPool::PoolThread::Run(void)
        {
            while (!owner_.IsStopping()) {
                ::node::sys::ThreadPool::Task *task = owner_.GetTask();
                if (task != NULL) {
                    task->task();
                    if (task->complete_callback) {
                        task->complete_callback();
                    }
                    owner_.FreeTask(task);
                }
            }
            owner_.task_queue_event_.Set();
        }
    }
}